import json
import time
import threading
from datetime import datetime, timedelta
from kafka import KafkaConsumer
from flask import Flask, jsonify, request, render_template
from flask_cors import CORS
import pandas as pd
import numpy as np
import webbrowser

# Kafka 配置信息
KAFKA_BOOTSTRAP_SERVERS = [
    "192.168.236.130:9092",
    "192.168.236.131:9092",
    "192.168.236.132:9092"
]
KAFKA_TOPIC = "stocks"
APP_PORT = 5000

# Flask 应用初始化
app = Flask(__name__, template_folder='templates', static_folder='static')
CORS(app)

# 全局数据存储
company_data = {}  # 最新数据
company_list = []  # 公司列表
industry_data = {"top_gainers": [], "top_market_cap": []}
kafka_connected = False
error_count = 0
last_update_time = 0

data_lock = threading.Lock()

# Kafka 消费线程
class KafkaStockConsumer(threading.Thread):
    def __init__(self):
        super().__init__(daemon=True)
        self.consumer = None
        self.running = True

    def run(self):
        global kafka_connected, company_data, company_list, industry_data, error_count, last_update_time
        while self.running:
            try:
                self.consumer = KafkaConsumer(
                    KAFKA_TOPIC,
                    bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
                    group_id='stock_visualization',
                    auto_offset_reset='latest',
                    value_deserializer=lambda m: m
                )
                kafka_connected = True
                print("Kafka消费者已连接")
                for message in self.consumer:
                    try:
                        records = json.loads(message.value.decode('utf-8'))
                        if isinstance(records, dict):
                            records = [records]
                        with data_lock:
                            for stock in records:
                                code = stock.get("code")
                                name = stock.get("name")
                                company_data[code] = stock
                                if not any(c["code"] == code for c in company_list):
                                    company_list.append({"code": code, "name": name})
                                    if len(company_list) > 200:
                                        company_list.pop(0)
                            # 定期更新行业数据
                            now = time.time()
                            if now - last_update_time > 10:
                                update_industry_data()
                                last_update_time = now
                    except Exception as e:
                        print(f"消息解析失败: {e}")
                kafka_connected = False
            except Exception as e:
                print(f"Kafka连接失败: {e}")
                kafka_connected = False
                error_count += 1
                time.sleep(5)

    def stop(self):
        self.running = False
        if self.consumer:
            self.consumer.close()

def update_industry_data():
    global industry_data
    if not company_data:
        industry_data = {"top_gainers": [], "top_market_cap": []}
        return
    df = pd.DataFrame(list(company_data.values()))
    if df.empty:
        industry_data = {"top_gainers": [], "top_market_cap": []}
        return
    # 涨幅前十
    top_gainers = df.nlargest(10, 'changepercent', keep='all')[['code', 'name', 'changepercent']].to_dict('records')
    # 市值前十
    top_market_cap = df.nlargest(10, 'mktcap', keep='all')[['code', 'name', 'mktcap']].to_dict('records')
    industry_data = {"top_gainers": top_gainers, "top_market_cap": top_market_cap}

# API 路由
@app.route("/api/connection-status", methods=["GET"])
def get_connection_status():
    return jsonify({"connected": kafka_connected, "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})

@app.route("/api/company-list", methods=["GET"])
def get_company_list():
    with data_lock:
        return jsonify({"success": True, "companies": company_list, "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})

@app.route("/api/realtime-data", methods=["GET"])
def get_realtime_data():
    company = request.args.get("company")
    if not company:
        return jsonify({"success": False, "message": "缺少公司代码参数"}), 400
    with data_lock:
        if company not in company_data:
            return jsonify({"success": False, "message": f"未找到公司代码为{company}的数据"}), 404
        return jsonify({"success": True, "data": company_data[company], "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})

@app.route("/api/industry-data", methods=["GET"])
def get_industry_data():
    with data_lock:
        return jsonify({"success": True, "data": industry_data, "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})

@app.route("/api/historical-data", methods=["GET"])
def get_historical_data():
    company = request.args.get("company")
    days = int(request.args.get("days", 30))
    if not company:
        return jsonify({"success": False, "message": "缺少公司代码参数"}), 400
    with data_lock:
        if company not in company_data:
            return jsonify({"success": False, "message": f"未找到公司代码为{company}的数据"}), 404
        # 模拟历史数据
        historical_data = []
        now = datetime.now()
        current_price = float(company_data[company].get("trade", 10))
        for i in range(days, 0, -1):
            date_str = (now - timedelta(days=i)).strftime("%Y-%m-%d")
            open_price = current_price * (1 + np.random.uniform(-0.03, 0.03))
            close_price = open_price * (1 + np.random.uniform(-0.02, 0.02))
            high_price = max(open_price, close_price) * (1 + np.random.uniform(0, 0.01))
            low_price = min(open_price, close_price) * (1 - np.random.uniform(0, 0.01))
            volume = int(np.random.uniform(800000, 1200000))
            historical_data.append({
                "date": date_str,
                "open": round(open_price, 2),
                "close": round(close_price, 2),
                "high": round(high_price, 2),
                "low": round(low_price, 2),
                "volume": volume
            })
        return jsonify({"success": True, "data": historical_data})

@app.route("/")
def index():
    return render_template("index.html")

if __name__ == "__main__":
    # 启动Kafka消费线程
    kafka_thread = KafkaStockConsumer()
    kafka_thread.start()
    # 自动打开首页
    webbrowser.open(f"http://localhost:{APP_PORT}/")
    app.run(host='0.0.0.0', port=APP_PORT, debug=True) 